package com.ks.winterwarmimage.mqtt.service;

import android.content.BroadcastReceiver;
import android.content.ComponentName;
import android.content.Context;
import android.content.Intent;
import android.content.IntentFilter;
import android.content.ServiceConnection;
import android.os.Bundle;
import android.os.IBinder;
import android.util.SparseArray;

import androidx.localbroadcastmanager.content.LocalBroadcastManager;

import org.eclipse.paho.client.mqttv3.DisconnectedBufferOptions;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.MqttToken;

import java.io.IOException;
import java.io.InputStream;
import java.security.KeyManagementException;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.cert.CertificateException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManager;
import javax.net.ssl.TrustManagerFactory;

public class MqttAndroidClient extends BroadcastReceiver implements IMqttAsyncClient {
    /**
     *
     * 从{@link MqttCallback#messageArrived(String, MqttMessage)}接收到的消息的确认模式
     *
     */
    public enum Ack {
        /**
         * 一旦{@link MqttCallback#messageArrived(String, MqttMessage)}返回，该消息即被确认为已收到。
         */
        AUTO_ACK,
        /**
         * 当{@link MqttCallback#messageArrived(String, MqttMessage)}返回时，
         * 消息将不会被确认为已收到，应用程序将不得不使用{@link MqttAndroidClient#acknowledgeMessage(String)}
         * 向{@link MqttAndroidClient}发出确认调用
         */
        MANUAL_ACK
    }

    private static final String SERVICE_NAME =MqttService.class.getName(); //"com.nursing.lamp.mqtt.service.MqttService";

    private static final int BIND_SERVICE_FLAG = 0;

    private static final ExecutorService pool = Executors.newCachedThreadPool();
    /**
     * ServiceConnection在绑定到服务时进行处理
     */
    private final class MyServiceConnection implements ServiceConnection {

        @Override
        public void onServiceConnected(ComponentName name, IBinder binder) {
            mqttService = ((MqttServiceBinder) binder).getService();
            bindedService = true;
            // now that we have the service available, we can actually
            // connect...
            doConnect();
        }

        @Override
        public void onServiceDisconnected(ComponentName name) {
            mqttService = null;
        }
    }
    // 服务连接或断开时的侦听器
    private final MyServiceConnection serviceConnection = new MyServiceConnection();

    // 将处理我们的mqtt调用的Android服务
    private MqttService mqttService;

    // 底层客户端连接的标识符，我们可以将其传递给服务
    private String clientHandle;

    private Context myContext;

    // 我们将各种令牌保存在一个集合中，并将它们的标识符传递给服务
    private final SparseArray<IMqttToken> tokenMap = new SparseArray<>();
    private int tokenNumber = 0;

    // 连接数据
    private final String serverURI;
    private final String clientId;
    private MqttClientPersistence persistence = null;
    private MqttConnectOptions connectOptions;
    private IMqttToken connectToken;

    // 应用程序提供的MqttCallback
    private MqttCallback callback;
    private MqttTraceHandler traceCallback;

    //应用程序已处理消息的确认
    private final Ack messageAck;
    private boolean traceEnabled = false;

    private volatile boolean receiverRegistered = false;
    private volatile boolean bindedService = false;
    /**
     * 构造函数 - 创建一个MqttAndroidClient，可用于与安卓上的MQTT服务器通信
     *
     * @param context 用于将上下文传递给回调的对象。
     * @param serverURI 指定用于连接到MQTT服务器的协议、主机名和端口
     * @param clientId 指定用于向服务器标识此连接的名称
     */
    public MqttAndroidClient(Context context, String serverURI, String clientId) {
        this(context, serverURI, clientId, null, Ack.AUTO_ACK);
    }
    /**
     * 构造函数 - 创建一个MqttAndroidClient，可用于与安卓上的MQTT服务器通信
     *
     * @param ctx 应用程序的上下文
     * @param serverURI 指定用于连接到MQTT服务器的协议、主机名和端口
     * @param clientId 指定用于向服务器标识此连接的名称
     * @param ackType 应用程序希望如何确认消息已被处理
     */
    public MqttAndroidClient(Context ctx, String serverURI, String clientId, Ack ackType) {
        this(ctx, serverURI, clientId, null, ackType);
    }
    /**
     * 构造函数 - 创建一个MqttAndroidClient，可用于与安卓上的MQTT服务器通信
     *
     * @param ctx 应用程序的上下文
     * @param serverURI 指定用于连接到MQTT服务器的协议、主机名和端口
     * @param clientId 指定用于向服务器标识此连接的名称
     * @param persistence 用于存储持久化数据的对象
     */
    public MqttAndroidClient(Context ctx, String serverURI, String clientId, MqttClientPersistence persistence) {
        this(ctx, serverURI, clientId, persistence, Ack.AUTO_ACK);
    }
    /**
     * 构造函数 - 创建一个MqttAndroidClient，可用于与安卓上的MQTT服务器通信
     *
     * @param context 用于将上下文传递给回调。
     * @param serverURI 指定用于连接到MQTT服务器的协议、主机名和端口
     * @param clientId 指定用于向服务器标识此连接的名称
     * @param persistence 用于存储飞行中消息的持久化类。如果为null，则使用默认的持久性机制
     * @param ackType 应用程序希望如何确认消息已被处理.
     */
    public MqttAndroidClient(Context context, String serverURI,
                             String clientId, MqttClientPersistence persistence, Ack ackType) {
        myContext = context;
        this.serverURI = serverURI;
        this.clientId = clientId;
        this.persistence = persistence;
        messageAck = ackType;
    }
    /**
     * 确定此客户端当前是否连接到服务器。
     *
     * @return <code>true</code> 连接, <code>false</code> 其它.
     */
    @Override
    public boolean isConnected() {
        return clientHandle != null && mqttService != null && mqttService.isConnected(clientHandle);
    }
    /**
     * 返回此客户端使用的客户端ID。
     * <p>
     * 连接到同一服务器或服务器场的所有客户端都必须具有唯一的ID。
     * </p>
     *
     * @return 此客户端使用的客户端ID。
     */
    @Override
    public String getClientId() {
        return clientId;
    }

    /**
     * 返回此客户端使用的服务器的URI地址。
     * <p>
     * 返回的String的格式与构造函数上使用的格式相同。
     * </p>
     *
     * @return 服务器的地址，作为URI字符串。
     */
    @Override
    public String getServerURI() {
        return serverURI;
    }

    /**
     * 关闭客户端。释放与客户端关联的所有资源。客户端关闭后，无法重新使用。例如，连接尝试将失败。
     *
     */
    @Override
    public void close() {
        if(mqttService != null){
            if (clientHandle == null) {
                clientHandle = mqttService.getClient(serverURI, clientId, myContext.getApplicationInfo().packageName,persistence);
            }
            mqttService.close(clientHandle);
        }
    }

    /**
     * 使用默认选项连接到MQTT服务器。
     * <p>
     * 默认选项在{@link MqttConnectOptions}类中指定。
     * </p>
     *
     * @throws MqttException
     *             对于任何相关问题
     * @return 用于跟踪和等待连接完成的令牌。如果设置了回调，则令牌将传递给回调方法。
     * @see #connect(MqttConnectOptions, Object, IMqttActionListener)
     */
    @Override
    public IMqttToken connect() throws MqttException {
        return connect(null, null);
    }


    /**
     * 使用提供的连接选项连接到MQTT服务器。
     * <p>
     * 将使用｛@link MqttConnectOptions｝参数中指定的选项建立连接。
     * </p>
     *
     * @param options
     *            一组覆盖默认值的连接参数。
     * @throws MqttException
     *             对于任何相关问题
     * @return 用于跟踪和等待连接完成的令牌。 令牌将传递给已设置的任何回调。
     * @see #connect(MqttConnectOptions, Object, IMqttActionListener)
     */
    @Override
    public IMqttToken connect(MqttConnectOptions options) throws MqttException {
        return connect(options, null, null);
    }

    /**
     * 使用默认选项连接到MQTT服务器。
     * <p>
     * 默认选项在{@link MqttConnectOptions}类中指定。
     * </p>
     *
     * @param userContext
     *            用于向回调传递上下文的可选对象。如果不需要，请使用null。
     * @param callback
     *            连接完成时将通知的可选侦听器。如果不需要，请使用null。
     * @throws MqttException
     *             对于任何相关问题
     * @return 用于跟踪和等待连接完成的令牌。 令牌将传递给已设置的任何回调。
     * @see #connect(MqttConnectOptions, Object, IMqttActionListener)
     */
    @Override
    public IMqttToken connect(Object userContext, IMqttActionListener callback)
            throws MqttException {
        return connect(new MqttConnectOptions(), userContext, callback);
    }

    /**
     * 使用指定的选项连接到MQTT服务器。
     * <p>
     * 构造函数上指定了要连接的服务器。 It is
     * 建议在连接之前调用{@link #setCallback(MqttCallback)}，以便在客户端连接后立即接受发往客户端的消息。
     * </p>
     *
     * <p>
     * 该方法在连接完成之前返回控制权。完成情况可以通过以下方式跟踪:
     * </p>
     * <ul>
     * <li>正在等待返回的令牌{@link IMqttToken#waitForCompletion()}
     * or</li>
     * <li>传入回调{@link IMqttActionListener}</li>
     * </ul>
     *
     *
     * @param options
     *            一组覆盖默认值的连接参数。
     * @param userContext
     *            用于将上下文传递给回调的可选对象。如果不需要，请使用null。
     * @param callback
     *            连接完成时将通知的可选侦听器。如果不需要，请使用null。
     * @return 用于跟踪和等待连接完成的令牌。 令牌将传递给已设置的任何回调。
     * @throws MqttException
     *             对于任何相关问题, 包括通信错误
     */

    @Override
    public IMqttToken connect(MqttConnectOptions options, Object userContext,
                              IMqttActionListener callback) throws MqttException {

        IMqttToken token = new MqttTokenAndroid(this, userContext,callback);

        connectOptions = options;
        connectToken = token;

        /*
         * 实际的连接取决于服务，我们在这里启动并绑定到该服务，但在serviceConnection onServiceConnected()
         * 方法异步运行之前，我们无法实际使用该服务，因此连接本身发生在onServiceConnected()方法中
         */
        if (mqttService == null) { // 第一次-必须绑定到服务
            Intent serviceStartIntent = new Intent();
            serviceStartIntent.setClassName(myContext, SERVICE_NAME);
            Object service = myContext.startService(serviceStartIntent);
            if (service == null) {
                IMqttActionListener listener = token.getActionCallback();
                if (listener != null) {
                    listener.onFailure(token, new RuntimeException("cannot start service " + SERVICE_NAME));
                }
            }

            //我们绑定BIND_SERVICE_FLAG (0)，让我们管理生命周期，直到最后一次调用stopService()停止它
            myContext.bindService(serviceStartIntent, serviceConnection,Context.BIND_AUTO_CREATE);

            if (!receiverRegistered) registerReceiver(this);
        }
        else {
            pool.execute(() -> {
                doConnect();

                //注册接收器以显示接收者。
                if (!receiverRegistered) registerReceiver(MqttAndroidClient.this);
            });
        }

        return token;
    }

    private void registerReceiver(BroadcastReceiver receiver) {
        IntentFilter filter = new IntentFilter();
        filter.addAction(MqttServiceConstants.CALLBACK_TO_ACTIVITY);
        LocalBroadcastManager.getInstance(myContext).registerReceiver(receiver, filter);
        receiverRegistered = true;
    }

    /**
     * 实际执行mqtt连接操作
     */
    private void doConnect() {
        if (clientHandle == null) {
            clientHandle = mqttService.getClient(serverURI, clientId,myContext.getApplicationInfo().packageName,
                    persistence);
        }
        mqttService.setTraceEnabled(traceEnabled);
        mqttService.setTraceCallbackId(clientHandle);

        String activityToken = storeToken(connectToken);
        try {
            mqttService.connect(clientHandle, connectOptions, null,
                    activityToken);
        }
        catch (MqttException e) {
            IMqttActionListener listener = connectToken.getActionCallback();
            if (listener != null) {
                listener.onFailure(connectToken, e);
            }
        }
    }

    /**
     * 断开与服务器的连接。
     * <p>
     * 尝试使客户端停顿，以便在断开连接之前完成未完成的工作。在断开连接之前，它将等待最多30秒的工作停止。
     * 不得从{@link MqttCallback}方法内部调用此方法。
     * </p>
     *
     * @return 用于跟踪和等待断开连接完成的令牌。 令牌将传递给已设置的任何回调。
     * @throws MqttException
     *             用于断开连接时遇到的问题
     * @see #disconnect(long, Object, IMqttActionListener)
     */
    @Override
    public IMqttToken disconnect() throws MqttException {
        IMqttToken token = new MqttTokenAndroid(this, null,
                null);
        String activityToken = storeToken(token);
        mqttService.disconnect(clientHandle, null, activityToken);
        return token;
    }

    /**
     * 断开与服务器的连接。
     * <p>
     * 尝试使客户端停顿，以便在断开连接之前完成未完成的工作。
     * 在断开连接之前，它将等待指定的最长停顿时间以完成工作。
     * 不得从{@link MqttCallback}方法内部调用此方法。
     * </p>
     *
     * @param quiesceTimeout
     *            在断开连接之前，允许现有工作完成的时间量（以毫秒为单位）。
     *            值为零或更小意味着客户端不会停止。
     * @return 用于跟踪和等待断开连接完成的令牌。 如果设置了回调，则令牌将传递给回调方法。
     * @throws MqttException
     *             用于断开连接时遇到的问题
     * @see #disconnect(long, Object, IMqttActionListener)
     */
    @Override
    public IMqttToken disconnect(long quiesceTimeout) throws MqttException {
        IMqttToken token = new MqttTokenAndroid(this, null,
                null);
        String activityToken = storeToken(token);
        mqttService.disconnect(clientHandle, quiesceTimeout, null,
                activityToken);
        return token;
    }

    /**
     * 断开与服务器的连接。
     * <p>
     * 尝试使客户端停顿，以便在断开连接之前完成未完成的工作。在断开连接之前，它将等待最多30秒的工作停止。
     * 不得从{@link MqttCallback}方法内部调用此方法。
     * </p>
     *
     * @param userContext
     *            用于向回调传递上下文的可选对象。如果不需要，请使用null。
     * @param callback
     *            可选的侦听器，当断开连接完成时将收到通知。如果不需要，请使用null。
     * @return 用于跟踪和等待断开连接完成的令牌。 令牌将传递给已设置的任何回调。
     * @throws MqttException
     *             用于断开连接时遇到的问题
     * @see #disconnect(long, Object, IMqttActionListener)
     */
    @Override
    public IMqttToken disconnect(Object userContext,
                                 IMqttActionListener callback) throws MqttException {
        IMqttToken token = new MqttTokenAndroid(this, userContext,
                callback);
        String activityToken = storeToken(token);
        mqttService.disconnect(clientHandle, null, activityToken);
        return token;
    }

    /**
     * 断开与服务器的连接。
     * <p>
     * 客户端将等待{@link MqttCallback}方法完成。 然后，它将等待暂停超时，以便完成已启动的工作。
     * 例如，当QoS 2消息已经开始流向服务器，但QoS 2流尚未完成时。它阻止新消息被接受，并且不会将任
     * 何已被接受但尚未开始通过网络传递的消息发送到服务器。当工作完成或暂停超时后，客户端将与服务器断开连接。
     * 如果cleanSession标志设置为false，并且下次在连接中也设置为false的话，那么在QoS 1或2中以前未传递的消息这次将被传递。
     * </p>
     * <p>
     * 不得从{@link MqttCallback}方法内部调用此方法。
     * </p>
     * <p>
     * 该方法在断开连接完成之前返回控制权。完成情况可以通过以下方式跟踪:
     * </p>
     * <ul>
     * <li>正在等待返回的令牌{@link IMqttToken#waitForCompletion()}
     * or</li>
     * <li>传入回调 {@link IMqttActionListener}</li>
     * </ul>
     *
     * @param quiesceTimeout
     *            在断开连接之前，允许现有工作完成的时间量（以毫秒为单位）。
     *            值为零或更小意味着客户端不会停止。
     * @param userContext
     *            用于向回调传递上下文的可选对象。如果不需要，请使用null。
     * @param callback
     *            可选的侦听器，当断开连接完成时将收到通知。如果不需要，请使用null。
     * @return 用于跟踪和等待断开连接完成的令牌。 令牌将传递给已设置的任何回调。
     * @throws MqttException
     *             用于断开连接时遇到的问题
     */
    @Override
    public IMqttToken disconnect(long quiesceTimeout, Object userContext,
                                 IMqttActionListener callback) throws MqttException {
        IMqttToken token = new MqttTokenAndroid(this, userContext,
                callback);
        String activityToken = storeToken(token);
        mqttService.disconnect(clientHandle, quiesceTimeout, null,
                activityToken);
        return token;
    }

    /**
     * 将消息发布到服务器上的主题。
     * <p>
     * 一种方便的方法，它将使用字节数组有效载荷和指定的QoS创建一个新的{@link MqttMessage}对象，然后发布它。
     * </p>
     *
     * @param topic
     *            将消息传递给，例如“finance/stock/ibm”。
     * @param payload
     *            用作有效载荷的字节数组
     * @param qos
     *            传递消息的服务质量。有效值为0、1或2。
     * @param retained
     *            服务器是否应该保留此消息。
     * @return 用于跟踪和等待发布完成的令牌。令牌将传递给已设置的任何回调。
     * @throws MqttPersistenceException
     *             当存储消息时出现问题
     * @throws IllegalArgumentException
     *             如果QoS的值不是0、1或2。
     * @throws MqttException
     *             发布消息时遇到的其他错误。
     *             例如，正在处理的消息太多。
     * @see #publish(String, MqttMessage, Object, IMqttActionListener)
     */
    @Override
    public IMqttDeliveryToken publish(String topic, byte[] payload, int qos,
                                      boolean retained) throws MqttException, MqttPersistenceException {
        return publish(topic, payload, qos, retained, null, null);
    }

    /**
     * 将消息发布到服务器上的主题。 接收{@link MqttMessage}消息，并按照请求的服务质量将其传递到服务器。
     *
     * @param topic
     *            将消息传递给，例如“finance/stock/ibm”。
     * @param message
     *            交付到服务器
     * @return 用于跟踪和等待发布完成的令牌。令牌将传递给已设置的任何回调。
     * @throws MqttPersistenceException
     *             当存储消息时出现问题
     * @throws IllegalArgumentException
     *             如果QoS的值不是0、1或2。
     * @throws MqttException
     *             发布消息时遇到的其他错误。
     *             例如，客户端未连接。
     * @see #publish(String, MqttMessage, Object, IMqttActionListener)
     */
    @Override
    public IMqttDeliveryToken publish(String topic, MqttMessage message)
            throws MqttException, MqttPersistenceException {
        return publish(topic, message, null, null);
    }

    /**
     * 将消息发布到服务器上的主题。
     * <p>
     * 一种方便的方法，它将创建一个新的{@link MqttMessage}对象，该对象具有字节数组有效载荷、指定的QoS和保留，然后将其发布。
     * </p>
     *
     * @param topic
     *            将消息传递给，例如“finance/stock/ibm”。
     * @param payload
     *            用作有效载荷的字节数组
     * @param qos
     *            传递消息的服务质量。有效值为0、1或2。
     * @param retained
     *            服务器是否应该保留此消息。
     * @param userContext
     *            用于向回调传递上下文的可选对象。如果不需要，请使用null。
     * @param callback
     *            可选侦听器，当消息传递完成到请求的服务质量时将收到通知
     * @return 用于跟踪和等待发布完成的令牌。令牌将传递给已设置的任何回调。
     * @throws MqttPersistenceException
     *             当存储消息时出现问题
     * @throws IllegalArgumentException
     *             如果QoS的值不是0、1或2。
     * @throws MqttException
     *             发布消息时遇到的其他错误。
     *             例如，客户端未连接。
     * @see #publish(String, MqttMessage, Object, IMqttActionListener)
     */
    @Override
    public IMqttDeliveryToken publish(String topic, byte[] payload, int qos,
                                      boolean retained, Object userContext, IMqttActionListener callback)
            throws MqttException, MqttPersistenceException {

        MqttMessage message = new MqttMessage(payload);
        message.setQos(qos);
        message.setRetained(retained);
        MqttDeliveryTokenAndroid token = new MqttDeliveryTokenAndroid(
                this, userContext, callback, message);
        String activityToken = storeToken(token);
        IMqttDeliveryToken internalToken = mqttService.publish(clientHandle,
                topic, payload, qos, retained, null, activityToken);
        token.setDelegate(internalToken);
        return token;
    }

    /**
     * 将消息发布到服务器上的主题。
     * <p>
     * 一旦此方法干净地返回，消息就被客户端接受发布，并将在后台线程上传递。
     * 如果连接失败或客户端停止，则一旦与服务器重新建立连接，消息将按请求的服务质量传递，条件是:
     * </p>
     * <ul>
     * <li>使用相同的客户端ID重新建立连接
     * <li>原始连接是在将{@link MqttConnectOptions#setCleanSession(boolean)}设置为false的情况下建立的
     * <li>将{@link MqttConnectOptions#setCleanSession(boolean)}设置为false，重新建立连接
     * <li>根据故障发生的时间，QoS 0消息可能无法传递。
     * </ul>
     *
     * <p>
     * 在构建应用程序时，主题树的设计应考虑以下主题名称语法和语义原则:
     * </p>
     *
     * <ul>
     * <li>主题必须至少有一个字符长。</li>
     * <li>主题名称区分大小写。例如，<em>ACCOUNTS</em>和<em>Accounts</em>是两个不同的主题。</li>
     * <li>主题名称可以包含空格字符。例如，
     * <em>Accounts
     * 	payable</em>是一个有效的主题</li>
     * <li>前导的“/”创建了一个特殊的主题。例如，
     * <em>/finance</em>不同于<em>finance</em>。 <em>/finance</em>
     * 匹配 "+/+"和"/+", 但不匹配 "+".</li>
     * <li>不要在任何主题中包含空字符(Unicode <em>\x0000</em>) 。</li>
     * </ul>
     *
     * <p>
     * 以下原则适用于主题树的构造和内容：
     * </p>
     *
     * <ul>
     * <li>长度限制为64k，但在此范围内，主题树中的级别数量没有限制。</li>
     * <li>可以有任意数量的根节点；也就是说，可以有任意数量的主题树。</li>
     * </ul>
     * <p>
     * 该方法在发布完成之前返回控制权。可通过以下方式跟踪完成情况：
     * </p>
     * <ul>
     * <li>设置一个{@link IMqttAsyncClient#setCallback(MqttCallback)}，
     * 其中将调用{@link MqttCallback#deliveryComplete(IMqttDeliveryToken)}方法。</li>
     * <li>等待返回的令牌{@link MqttToken#waitForCompletion()}或</li>
     * <li>将回调{@link IMqttActionListener}传递给此方法</li>
     * </ul>
     *
     * @param topic 将消息传递给，例如“finance/stock/ibm”。
     * @param message 交付到服务器
     * @param userContext  用于向回调传递上下文的可选对象。如果不需要，请使用null。
     * @param callback 可选侦听器，当消息传递完成到请求的服务质量时将收到通知
     * @return 用于跟踪和等待发布完成的令牌。如果设置了令牌，它将被传递给回调方法。
     * @throws MqttPersistenceException 当存储消息时出现问题
     * @throws IllegalArgumentException 如果QoS的值不是0、1或2。
     * @throws MqttException 发布消息时遇到的其他错误。 例如，客户端未连接。
     * @see MqttMessage
     */
    @Override
    public IMqttDeliveryToken publish(String topic, MqttMessage message,
                                      Object userContext, IMqttActionListener callback)
            throws MqttException, MqttPersistenceException {
        MqttDeliveryTokenAndroid token = new MqttDeliveryTokenAndroid(
                this, userContext, callback, message);
        String activityToken = storeToken(token);
        IMqttDeliveryToken internalToken = mqttService.publish(clientHandle,
                topic, message, null, activityToken);
        token.setDelegate(internalToken);
        return token;
    }

    /**
     * 订阅一个主题，其中可能包含通配符。
     *
     * @param topic
     *            要订阅的主题，其中可以包括通配符。
     * @param qos
     *            订阅的最高服务质量。 Messages
     *            published at a lower quality of service will be received at
     *            the published QoS. Messages published at a higher quality of
     *            service will be received using the QoS specified on the
     *            subscription.
     * @return token used to track and wait for the subscribe to complete. 如果设置了令牌，它将被传递给回调方法。
     * @throws MqttSecurityException
     *             for security related problems
     * @throws MqttException
     *             for non security related problems
     *
     * @see #subscribe(String[], int[], Object, IMqttActionListener)
     */
    @Override
    public IMqttToken subscribe(String topic, int qos) throws MqttException,
            MqttSecurityException {
        return subscribe(topic, qos, null, null);
    }

    /**
     * Subscribe to multiple topics, each topic may include wildcards.
     *
     * <p>
     * Provides an optimized way to subscribe to multiple topics compared to
     * subscribing to each one individually.
     * </p>
     *
     * @param topic
     *            one or more topics to subscribe to, which can include
     *            wildcards
     * @param qos
     *            订阅的最高服务质量。 Messages
     *            published at a lower quality of service will be received at
     *            the published QoS. Messages published at a higher quality of
     *            service will be received using the QoS specified on the
     *            subscription.
     * @return token used to track and wait for the subscription to complete. 如果设置了令牌，它将被传递给回调方法。
     * @throws MqttSecurityException
     *             for security related problems
     * @throws MqttException
     *             for non security related problems
     *
     * @see #subscribe(String[], int[], Object, IMqttActionListener)
     */
    @Override
    public IMqttToken subscribe(String[] topic, int[] qos)
            throws MqttException, MqttSecurityException {
        return subscribe(topic, qos, null, null);
    }

    /**
     * 订阅一个主题，其中可能包含通配符。
     *
     * @param topic
     *            要订阅的主题，其中可以包括通配符。
     * @param qos
     *            订阅的最高服务质量。 Messages
     *            published at a lower quality of service will be received at
     *            the published QoS. Messages published at a higher quality of
     *            service will be received using the QoS specified on the
     *            subscription.
     * @param userContext
     *            用于向回调传递上下文的可选对象。如果不需要，请使用null。
     * @param callback
     *            optional listener that will be notified when subscribe has
     *            completed
     * @return token used to track and wait for the subscribe to complete. 如果设置了令牌，它将被传递给回调方法。
     * @throws MqttException
     *             if there was an error when registering the subscription.
     *
     * @see #subscribe(String[], int[], Object, IMqttActionListener)
     */
    @Override
    public IMqttToken subscribe(String topic, int qos, Object userContext,
                                IMqttActionListener callback) throws MqttException {
        IMqttToken token = new MqttTokenAndroid(this, userContext,
                callback, new String[]{topic});
        String activityToken = storeToken(token);
        mqttService.subscribe(clientHandle, topic, qos, null, activityToken);
        return token;
    }

    /**
     * Subscribes to multiple topics, each topic may include wildcards.
     * <p>
     * Provides an optimized way to subscribe to multiple topics compared to
     * subscribing to each one individually.
     * </p>
     * <p>
     * The {@link #setCallback(MqttCallback)} method should be called before
     * this method, otherwise any received messages will be discarded.
     * </p>
     * <p>
     * If (@link MqttConnectOptions#setCleanSession(boolean)} was set to true,
     * when connecting to the server, the subscription remains in place until
     * either:
     * </p>
     * <ul>
     * <li>The client disconnects</li>
     * <li>An unsubscribe method is called to unsubscribe the topic</li>
     * </ul>
     * <p>
     * If (@link MqttConnectOptions#setCleanSession(boolean)} was set to false,
     * when connecting to the server, the subscription remains in place
     * until either:
     * </p>
     * <ul>
     * <li>An unsubscribe method is called to unsubscribe the topic</li>
     * <li>The next time the client connects with cleanSession set to true
     * </ul>
     * <p>With cleanSession set to false the MQTT server will store messages
     * on behalf of the client when the client is not connected. The next time
     * the client connects with the <b>same client ID</b> the server will
     * deliver the stored messages to the client.
     * </p>
     *
     * <p>
     * The "topic filter" string is used when subscription may contain special
     * characters, which allows you to subscribe to multiple topics at once.
     * <dl>
     * <dt>Topic level separator</dt>
     * <dd>The forward slash (/) is used to separate each level within a topic
     * tree and provide a hierarchical structure to the topic space. The use of
     * the topic level separator is significant when the two wildcard characters
     * are encountered in topics specified by subscribers.</dd>
     *
     * <dt>Multi-level wildcard</dt>
     * <dd>
     * <p>
     * The number sign (#) is a wildcard character that matches any number of
     * levels within a topic. For example, if you subscribe to <span><span
     * class="filepath">finance/stock/ibm/#</span></span>, you receive messages
     * on these topics:
     * </p>
     * <ul>
     *     <li><pre>finance/stock/ibm</pre></li>
     *     <li><pre>finance/stock/ibm/closingprice</pre></li>
     *     <li><pre>finance/stock/ibm/currentprice</pre></li>
     * </ul>
     *
     * <p>
     * The multi-level wildcard can represent zero or more levels. Therefore,
     * <em>finance/#</em> can also match the singular <em>finance</em>, where
     * <em>#</em> represents zero levels. The topic level separator is
     * meaningless in this context, because there are no levels to separate.
     * </p>
     *
     * <p>
     * The <span>multi-level</span> wildcard can be specified only on its own or
     * next to the topic level separator character. Therefore, <em>#</em> and
     * <em>finance/#</em> are both valid, but <em>finance#</em> is not valid.
     * <span>The multi-level wildcard must be the last character used within the
     * topic tree. For example, <em>finance/#</em> is valid but
     * <em>finance/#/closingprice</em> is not valid.</span>
     * </p>
     * </dd>
     *
     * <dt>Single-level wildcard</dt>
     * <dd>
     * <p>
     * The plus sign (+) is a wildcard character that matches only one topic
     * level. For example, <em>finance/stock/+</em> matches
     * <em>finance/stock/ibm</em> and <em>finance/stock/xyz</em>, but not
     * <em>finance/stock/ibm/closingprice</em>. Also, because the single-level
     * wildcard matches only a single level, <em>finance/+</em> does not match
     * <em>finance</em>.
     * </p>
     *
     * <p>
     * Use the single-level wildcard at any level in the topic tree, and in
     * conjunction with the multilevel wildcard. Specify the single-level
     * wildcard next to the topic level separator, except when it is specified
     * on its own. Therefore, <em>+</em> and <em>finance/+</em> are both valid,
     * but <em>finance+</em> is not valid. <span>The single-level wildcard can
     * be used at the end of the topic tree or within the topic tree. For
     * example, <em>finance/+</em> and <em>finance/+/ibm</em> are both
     * valid.</span>
     * </p>
     * </dd>
     * </dl>
     * <p>
     * The method returns control before the subscribe completes. Completion can
     * be tracked by:
     * </p>
     * <ul>
     * <li>Waiting on the supplied token {@link MqttToken#waitForCompletion()}
     * or</li>
     * <li>Passing in a callback {@link IMqttActionListener} to this method</li>
     * </ul>
     *
     * @param topic
     *            one or more topics to subscribe to, which can include
     *            wildcards
     * @param qos
     *            the maximum quality of service to subscribe each topic
     *            at.Messages published at a lower quality of service will be
     *            received at the published QoS. Messages published at a higher
     *            quality of service will be received using the QoS specified on
     *            the subscription.
     * @param userContext
     *            用于向回调传递上下文的可选对象。如果不需要，请使用null。
     * @param callback
     *            optional listener that will be notified when subscribe has
     *            completed
     * @return token used to track and wait for the subscribe to complete. 如果设置了令牌，它将被传递给回调方法。
     * @throws MqttException
     *             if there was an error registering the subscription.
     * @throws IllegalArgumentException
     *             if the two supplied arrays are not the same size.
     */
    @Override
    public IMqttToken subscribe(String[] topic, int[] qos, Object userContext,
                                IMqttActionListener callback) throws MqttException {
        IMqttToken token = new MqttTokenAndroid(this, userContext,
                callback, topic);
        String activityToken = storeToken(token);
        mqttService.subscribe(clientHandle, topic, qos, null, activityToken);
        return token;
    }

    /**
     * 订阅一个主题，其中可能包含通配符。
     *
     * @see #subscribe(String[], int[], Object, IMqttActionListener)
     *
     * @param topicFilter 要订阅的主题，其中可以包括通配符。
     * @param qos 订阅的最高服务质量。 Messages
     * published at a lower quality of service will be received at the published
     * QoS.  Messages published at a higher quality of service will be received using
     * the QoS specified on the subscribe.
     * @param userContext optional object used to pass context to the callback. Use
     * null if not required.
     * @param callback optional listener that will be notified when subscribe
     * has completed
     * @param messageListener a callback to handle incoming messages
     * @return token used to track and wait for the subscribe to complete. The token
     * will be passed to callback methods if set.
     * @throws MqttException if there was an error registering the subscription.
     */
    public IMqttToken subscribe(String topicFilter, int qos, Object userContext, IMqttActionListener callback, IMqttMessageListener messageListener) throws MqttException {

        return subscribe(new String[] {topicFilter}, new int[] {qos}, userContext, callback, new IMqttMessageListener[] {messageListener});
    }

    /**
     * 订阅一个主题，其中可能包含通配符。
     *
     * @see #subscribe(String[], int[], Object, IMqttActionListener)
     *
     * @param topicFilter 要订阅的主题，其中可以包括通配符。
     * @param qos 订阅的最高服务质量。 Messages
     * published at a lower quality of service will be received at the published
     * QoS.  Messages published at a higher quality of service will be received using
     * the QoS specified on the subscribe.
     * @param messageListener a callback to handle incoming messages
     * @return token used to track and wait for the subscribe to complete. The token
     * will be passed to callback methods if set.
     * @throws MqttException if there was an error registering the subscription.
     */
    public IMqttToken subscribe(String topicFilter, int qos, IMqttMessageListener messageListener) throws MqttException {

        return subscribe(topicFilter, qos, null, null, messageListener);
    }


    /**
     * Subscribe to multiple topics, each of which may include wildcards.
     *
     * <p>Provides an optimized way to subscribe to multiple topics compared to
     * subscribing to each one individually.</p>
     *
     * @see #subscribe(String[], int[], Object, IMqttActionListener)
     *
     * @param topicFilters one or more topics to subscribe to, which can include wildcards
     * @param qos 订阅的最高服务质量。 Messages
     * published at a lower quality of service will be received at the published
     * QoS.  Messages published at a higher quality of service will be received using
     * the QoS specified on the subscribe.
     * @param messageListeners an array of callbacks to handle incoming messages
     * @return token used to track and wait for the subscribe to complete. The token
     * will be passed to callback methods if set.
     * @throws MqttException if there was an error registering the subscription.
     */
    public IMqttToken subscribe(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners) throws MqttException {

        return subscribe(topicFilters, qos, null, null, messageListeners);
    }


    /**
     * Subscribe to multiple topics, each of which may include wildcards.
     *
     * <p>Provides an optimized way to subscribe to multiple topics compared to
     * subscribing to each one individually.</p>
     *
     * @see #subscribe(String[], int[], Object, IMqttActionListener)
     *
     * @param topicFilters one or more topics to subscribe to, which can include wildcards
     * @param qos 订阅的最高服务质量。 Messages
     * published at a lower quality of service will be received at the published
     * QoS.  Messages published at a higher quality of service will be received using
     * the QoS specified on the subscribe.
     * @param userContext optional object used to pass context to the callback. Use
     * null if not required.
     * @param callback optional listener that will be notified when subscribe
     * has completed
     * @param messageListeners an array of callbacks to handle incoming messages
     * @return token used to track and wait for the subscribe to complete. The token
     * will be passed to callback methods if set.
     * @throws MqttException if there was an error registering the subscription.
     */
    public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext, IMqttActionListener callback, IMqttMessageListener[] messageListeners) throws MqttException {
        IMqttToken token = new MqttTokenAndroid(this, userContext, callback, topicFilters);
        String activityToken = storeToken(token);
        mqttService.subscribe(clientHandle, topicFilters, qos, null, activityToken, messageListeners);

        return null;
    }


    /**
     * Requests the server unsubscribe the client from a topic.
     *
     * @param topic
     *            the topic to unsubscribe from. It must match a topic specified
     *            on an earlier subscribe.
     * @return 用于跟踪和等待取消订阅完成的令牌。 如果设置了令牌，它将被传递给回调方法。
     * @throws MqttException
     *             if there was an error unregistering the subscription.
     *
     * @see #unsubscribe(String[], Object, IMqttActionListener)
     */
    @Override
    public IMqttToken unsubscribe(String topic) throws MqttException {
        return unsubscribe(topic, null, null);
    }

    /**
     * Requests the server to unsubscribe the client from one or more topics.
     *
     * @param topic
     *            one or more topics to unsubscribe from. Each topic must match
     *            one specified on an earlier subscription.
     * @return 用于跟踪和等待取消订阅完成的令牌。 如果设置了令牌，它将被传递给回调方法。
     * @throws MqttException
     *             if there was an error unregistering the subscription.
     *
     * @see #unsubscribe(String[], Object, IMqttActionListener)
     */
    @Override
    public IMqttToken unsubscribe(String[] topic) throws MqttException {
        return unsubscribe(topic, null, null);
    }

    /**
     * Requests the server to unsubscribe the client from a topics.
     *
     * @param topic
     *            the topic to unsubscribe from. It must match a topic specified
     *            on an earlier subscribe.
     * @param userContext
     *            用于向回调传递上下文的可选对象。如果不需要，请使用null。
     * @param callback
     *            optional listener that will be notified when unsubscribe has
     *            completed
     * @return 用于跟踪和等待取消订阅完成的令牌。 如果设置了令牌，它将被传递给回调方法。
     * @throws MqttException
     *             if there was an error unregistering the subscription.
     *
     * @see #unsubscribe(String[], Object, IMqttActionListener)
     */
    @Override
    public IMqttToken unsubscribe(String topic, Object userContext,
                                  IMqttActionListener callback) throws MqttException {
        IMqttToken token = new MqttTokenAndroid(this, userContext,
                callback);
        String activityToken = storeToken(token);
        mqttService.unsubscribe(clientHandle, topic, null, activityToken);
        return token;
    }

    /**
     * Requests the server to unsubscribe the client from one or more topics.
     * <p>
     * Unsubcribing is the opposite of subscribing. When the server receives the
     * unsubscribe request it looks to see if it can find a matching
     * subscription for the client and then removes it. After this point the
     * server will send no more messages to the client for this subscription.
     * </p>
     * <p>
     * The topic(s) specified on the unsubscribe must match the topic(s)
     * specified in the original subscribe request for the unsubscribe to
     * succeed
     * </p>
     * <p>
     * The method returns control before the unsubscribe completes. Completion
     * can be tracked by:
     * </p>
     * <ul>
     * <li>Waiting on the returned token {@link MqttToken#waitForCompletion()}
     * or</li>
     * <li>Passing in a callback {@link IMqttActionListener} to this method</li>
     * </ul>
     *
     * @param topic
     *            one or more topics to unsubscribe from. Each topic must match
     *            one specified on an earlier subscription.
     * @param userContext
     *            用于向回调传递上下文的可选对象。如果不需要，请使用null。
     * @param callback
     *            optional listener that will be notified when unsubscribe has
     *            completed
     * @return 用于跟踪和等待取消订阅完成的令牌。 如果设置了令牌，它将被传递给回调方法。
     * @throws MqttException
     *             if there was an error unregistering the subscription.
     */
    @Override
    public IMqttToken unsubscribe(String[] topic, Object userContext,
                                  IMqttActionListener callback) throws MqttException {
        IMqttToken token = new MqttTokenAndroid(this, userContext,
                callback);
        String activityToken = storeToken(token);
        mqttService.unsubscribe(clientHandle, topic, null, activityToken);
        return token;
    }

    @Override
    public boolean removeMessage(IMqttDeliveryToken token) {
        return false;
    }

    /**
     * Returns the delivery tokens for any outstanding publish operations.
     * <p>
     * If a client has been restarted and there are messages that were in the
     * process of being delivered when the client stopped, this method returns a
     * token for each in-flight message to enable the delivery to be tracked.
     * Alternately the {@link MqttCallback#deliveryComplete(IMqttDeliveryToken)}
     * callback can be used to track the delivery of outstanding messages.
     * </p>
     * <p>
     * If a client connects with cleanSession true then there will be no
     * delivery tokens as the cleanSession option deletes all earlier state. For
     * state to be remembered the client must connect with cleanSession set to
     * false
     * </P>
     *
     * @return zero or more delivery tokens
     */
    @Override
    public IMqttDeliveryToken[] getPendingDeliveryTokens() {
        return mqttService.getPendingDeliveryTokens(clientHandle);
    }

    /**
     * Sets a callback listener to use for events that happen asynchronously.
     * <p>
     * There are a number of events that the listener will be notified about.
     * These include:
     * </p>
     * <ul>
     * <li>A new message has arrived and is ready to be processed</li>
     * <li>The connection to the server has been lost</li>
     * <li>Delivery of a message to the server has completed</li>
     * </ul>
     * <p>
     * Other events that track the progress of an individual operation such as
     * connect and subscribe can be tracked using the {@link MqttToken} returned
     * from each non-blocking method or using setting a
     * {@link IMqttActionListener} on the non-blocking method.
     * <p>
     *
     * @param callback
     *            which will be invoked for certain asynchronous events
     *
     * @see MqttCallback
     */
    @Override
    public void setCallback(MqttCallback callback) {
        this.callback = callback;

    }

    /**
     * identify the callback to be invoked when making tracing calls back into
     * the Activity
     *
     * @param traceCallback handler
     */
    public void setTraceCallback(MqttTraceHandler traceCallback) {
        this.traceCallback = traceCallback;
        // mqttService.setTraceCallbackId(traceCallbackId);
    }

    /**
     * turn tracing on and off
     *
     * @param traceEnabled
     *            set <code>true</code> to enable trace, otherwise, set
     *            <code>false</code> to disable trace
     *
     */
    public void setTraceEnabled(boolean traceEnabled) {
        this.traceEnabled = traceEnabled;
        if (mqttService !=null)
            mqttService.setTraceEnabled(traceEnabled);
    }

    /**
     * <p>
     * Process incoming Intent objects representing the results of operations
     * and asynchronous activities such as message received
     * </p>
     * <p>
     * <strong>Note:</strong> This is only a public method because the Android
     * APIs require such.<br>
     * This method should not be explicitly invoked.
     * </p>
     */
    @Override
    public void onReceive(Context context, Intent intent) {
        Bundle data = intent.getExtras();

        String handleFromIntent = data
                .getString(MqttServiceConstants.CALLBACK_CLIENT_HANDLE);

        if ((handleFromIntent == null)
                || (!handleFromIntent.equals(clientHandle))) {
            return;
        }

        String action = data.getString(MqttServiceConstants.CALLBACK_ACTION);

        if (MqttServiceConstants.CONNECT_ACTION.equals(action)) {
            connectAction(data);
        }
        else if (MqttServiceConstants.CONNECT_EXTENDED_ACTION.equals(action)){
            connectExtendedAction(data);
        }
        else if (MqttServiceConstants.MESSAGE_ARRIVED_ACTION.equals(action)) {
            messageArrivedAction(data);
        }
        else if (MqttServiceConstants.SUBSCRIBE_ACTION.equals(action)) {
            subscribeAction(data);
        }
        else if (MqttServiceConstants.UNSUBSCRIBE_ACTION.equals(action)) {
            unSubscribeAction(data);
        }
        else if (MqttServiceConstants.SEND_ACTION.equals(action)) {
            sendAction(data);
        }
        else if (MqttServiceConstants.MESSAGE_DELIVERED_ACTION.equals(action)) {
            messageDeliveredAction(data);
        }
        else if (MqttServiceConstants.ON_CONNECTION_LOST_ACTION
                .equals(action)) {
            connectionLostAction(data);
        }
        else if (MqttServiceConstants.DISCONNECT_ACTION.equals(action)) {
            disconnected(data);
        }
        else if (MqttServiceConstants.TRACE_ACTION.equals(action)) {
            traceAction(data);
        }else{
            mqttService.traceError(MqttService.TAG, "Callback action doesn't exist.");
        }

    }

    /**
     * Acknowledges a message received on the
     * {@link MqttCallback#messageArrived(String, MqttMessage)}
     *
     * @param messageId
     *            the messageId received from the MqttMessage (To access this
     *            field you need to cast {@link MqttMessage} to
     *            {@link ParcelableMqttMessage})
     * @return whether or not the message was successfully acknowledged
     */
    public boolean acknowledgeMessage(String messageId) {
        if (messageAck == Ack.MANUAL_ACK) {
            Status status = mqttService.acknowledgeMessageArrival(clientHandle, messageId);
            return status == Status.OK;
        }
        return false;

    }

    public void messageArrivedComplete(int messageId, int qos) {
        throw new UnsupportedOperationException();
    }

    public void setManualAcks(boolean manualAcks) {
        throw new UnsupportedOperationException();
    }

    @Override
    public void reconnect() throws MqttException {

    }

    /**
     * Process the results of a connection
     *
     * @param data
     */
    private void connectAction(Bundle data) {
        IMqttToken token = connectToken;
        removeMqttToken(data);

        simpleAction(token, data);
    }



    /**
     * Process a notification that we have disconnected
     *
     * @param data
     */
    private void disconnected(Bundle data) {
        clientHandle = null; // avoid reuse!
        IMqttToken token = removeMqttToken(data);
        if (token != null) {
            ((MqttTokenAndroid) token).notifyComplete();
        }
        if (callback != null) {
            callback.connectionLost(null);
        }
    }

    /**
     * Process a Connection Lost notification
     *
     * @param data
     */
    private void connectionLostAction(Bundle data) {
        if (callback != null) {
            Exception reason = (Exception) data
                    .getSerializable(MqttServiceConstants.CALLBACK_EXCEPTION);
            callback.connectionLost(reason);
        }
    }

    private void connectExtendedAction(Bundle data){
        // This is called differently from a normal connect

        if(callback instanceof MqttCallbackExtended){
            boolean reconnect = data.getBoolean(MqttServiceConstants.CALLBACK_RECONNECT, false);
            String serverURI = data.getString(MqttServiceConstants.CALLBACK_SERVER_URI);
            ((MqttCallbackExtended) callback).connectComplete(reconnect, serverURI);
        }

    }

    /**
     * Common processing for many notifications
     *
     * @param token
     *            the token associated with the action being undertake
     * @param data
     *            the result data
     */
    private void simpleAction(IMqttToken token, Bundle data) {
        if (token != null) {
            Status status = (Status) data
                    .getSerializable(MqttServiceConstants.CALLBACK_STATUS);
            if (status == Status.OK) {
                ((MqttTokenAndroid) token).notifyComplete();
            }
            else {
                Exception exceptionThrown = (Exception) data.getSerializable(MqttServiceConstants.CALLBACK_EXCEPTION);
                ((MqttTokenAndroid) token).notifyFailure(exceptionThrown);
            }
        } else {
            mqttService.traceError(MqttService.TAG, "simpleAction : token is null");
        }
    }

    /**
     * Process notification of a publish(send) operation
     *
     * @param data
     */
    private void sendAction(Bundle data) {
        IMqttToken token = getMqttToken(data); // get, don't remove - will
        // remove on delivery
        simpleAction(token, data);
    }

    /**
     * Process notification of a subscribe operation
     *
     * @param data
     */
    private void subscribeAction(Bundle data) {
        IMqttToken token = removeMqttToken(data);
        simpleAction(token, data);
    }

    /**
     * Process notification of an unsubscribe operation
     *
     * @param data
     */
    private void unSubscribeAction(Bundle data) {
        IMqttToken token = removeMqttToken(data);
        simpleAction(token, data);
    }

    /**
     * Process notification of a published message having been delivered
     *
     * @param data
     */
    private void messageDeliveredAction(Bundle data) {
        IMqttToken token = removeMqttToken(data);
        if (token != null) {
            if (callback != null) {
                Status status = (Status) data
                        .getSerializable(MqttServiceConstants.CALLBACK_STATUS);
                if (status == Status.OK && token instanceof IMqttDeliveryToken) {
                    callback.deliveryComplete((IMqttDeliveryToken) token);
                }
            }
        }
    }

    /**
     * Process notification of a message's arrival
     *
     * @param data
     */
    private void messageArrivedAction(Bundle data) {
        if (callback != null) {
            String messageId = data
                    .getString(MqttServiceConstants.CALLBACK_MESSAGE_ID);
            String destinationName = data
                    .getString(MqttServiceConstants.CALLBACK_DESTINATION_NAME);

            ParcelableMqttMessage message = data
                    .getParcelable(MqttServiceConstants.CALLBACK_MESSAGE_PARCEL);
            try {
                if (messageAck == Ack.AUTO_ACK) {
                    callback.messageArrived(destinationName, message);
                    mqttService.acknowledgeMessageArrival(clientHandle, messageId);
                }
                else {
                    message.messageId = messageId;
                    callback.messageArrived(destinationName, message);
                }

                // let the service discard the saved message details
            }
            catch (Exception e) {
                // Swallow the exception
            }
        }
    }

    /**
     * Process trace action - pass trace data back to the callback
     *
     * @param data
     */
    private void traceAction(Bundle data) {

        if (traceCallback != null) {
            String severity = data.getString(MqttServiceConstants.CALLBACK_TRACE_SEVERITY);
            String message =  data.getString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE);
            String tag = data.getString(MqttServiceConstants.CALLBACK_TRACE_TAG);
            if (MqttServiceConstants.TRACE_DEBUG.equals(severity))
                traceCallback.traceDebug(tag, message);
            else if (MqttServiceConstants.TRACE_ERROR.equals(severity))
                traceCallback.traceError(tag, message);
            else
            {
                Exception e = (Exception) data.getSerializable(MqttServiceConstants.CALLBACK_EXCEPTION);
                traceCallback.traceException(tag, message, e);
            }
        }
    }

    /**
     * @param token
     *            identifying an operation
     * @return an identifier for the token which can be passed to the Android
     *         Service
     */
    private synchronized String storeToken(IMqttToken token) {
        tokenMap.put(tokenNumber, token);
        return Integer.toString(tokenNumber++);
    }

    /**
     * Get a token identified by a string, and remove it from our map
     *
     * @param data
     * @return the token
     */
    private synchronized IMqttToken removeMqttToken(Bundle data) {

        String activityToken = data.getString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN);
        if (activityToken!=null){
            int tokenNumber = Integer.parseInt(activityToken);
            IMqttToken token = tokenMap.get(tokenNumber);
            tokenMap.delete(tokenNumber);
            return token;
        }
        return null;
    }

    /**
     * Get a token identified by a string, and remove it from our map
     *
     * @param data
     * @return the token
     */
    private synchronized IMqttToken getMqttToken(Bundle data) {
        String activityToken = data
                .getString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN);
        return tokenMap.get(Integer.parseInt(activityToken));
    }

    /**
     * Sets the DisconnectedBufferOptions for this client
     * @param bufferOpts the DisconnectedBufferOptions
     */
    public void setBufferOpts(DisconnectedBufferOptions bufferOpts) {
        mqttService.setBufferOpts(clientHandle, bufferOpts);
    }

    public int getBufferedMessageCount(){
        return mqttService.getBufferedMessageCount(clientHandle);
    }

    public MqttMessage getBufferedMessage(int bufferIndex){
        return mqttService.getBufferedMessage(clientHandle, bufferIndex);
    }

    public void deleteBufferedMessage(int bufferIndex){
        mqttService.deleteBufferedMessage(clientHandle, bufferIndex);
    }

    @Override
    public int getInFlightMessageCount() {
        return 0;
    }

    /**
     * Get the SSLSocketFactory using SSL key store and password
     * <p>
     * A convenience method, which will help user to create a SSLSocketFactory
     * object
     * </p>
     *
     * @param keyStore
     *            the SSL key store which is generated by some SSL key tool,
     *            such as keytool in Java JDK
     * @param password
     *            the password of the key store which is set when the key store
     *            is generated
     * @return SSLSocketFactory used to connect to the server with SSL
     *         authentication
     * @throws MqttSecurityException
     *             if there was any error when getting the SSLSocketFactory
     */
    public SSLSocketFactory getSSLSocketFactory (InputStream keyStore, String password) throws MqttSecurityException {
        try{
            SSLContext ctx = null;
            SSLSocketFactory sslSockFactory=null;
            KeyStore ts;
            ts = KeyStore.getInstance("BKS");
            ts.load(keyStore, password.toCharArray());
            TrustManagerFactory tmf = TrustManagerFactory.getInstance("X509");
            tmf.init(ts);
            TrustManager[] tm = tmf.getTrustManagers();
            ctx = SSLContext.getInstance("TLSv1");
            ctx.init(null, tm, null);

            sslSockFactory=ctx.getSocketFactory();
            return sslSockFactory;

        } catch (KeyStoreException | CertificateException | IOException | NoSuchAlgorithmException | KeyManagementException e) {
            throw new MqttSecurityException(e);
        }
    }

    @Override
    public void disconnectForcibly() {
        throw new UnsupportedOperationException();
    }

    @Override
    public void disconnectForcibly(long disconnectTimeout) {
        throw new UnsupportedOperationException();
    }

    @Override
    public void disconnectForcibly(long quiesceTimeout, long disconnectTimeout)
            throws MqttException {
        throw new UnsupportedOperationException();
    }

    /**
     * Unregister receiver which receives intent from MqttService avoids
     * IntentReceiver leaks.
     */
    public void unregisterResources(){
        if(myContext != null && receiverRegistered){
            synchronized (MqttAndroidClient.this) {
                LocalBroadcastManager.getInstance(myContext).unregisterReceiver(this);
                receiverRegistered = false;
            }
            if(bindedService){
                try{
                    myContext.unbindService(serviceConnection);
                    bindedService = false;
                }catch(IllegalArgumentException e){
                    //Ignore unbind issue.
                }
            }
        }
    }

    /**
     * Register receiver to receiver intent from MqttService. Call this method
     * when activity is hidden and become to show again.
     *
     * @param context
     *            - Current activity context.
     */
    public void registerResources(Context context){
        if(context != null){
            this.myContext = context;
            if(!receiverRegistered){
                registerReceiver(this);
            }
        }
    }
}
